package com.heima.article.send;

import com.alibaba.fastjson.JSON;
import com.heima.article.msg.ArticleBehaviorMsg;
import com.heima.common.constants.message.HotArticleConstants;
import com.heima.model.article.enums.ArticleBehaviorTypeEnum;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * kafka 消息生产者
 */
@Slf4j
public class ArticleBehaviorProducer {

    public static final String TOPIC = HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC;
    private static final String KAFKA_IP = "192.168.137.136:9092";



    public static void main(String[] args) throws InterruptedException {



        //添加kafka的配置信息
        Properties properties = new Properties();
        //配置broker信息
        //添加kafka的配置信息
        //配置broker信息
        properties.put("bootstrap.servers", KAFKA_IP);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.RETRIES_CONFIG, 10);

        //生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

        System.out.println("kafka Producer 启动");
        for (int i = 1; i <= 2; i++) {
            //发送消息
            try {
                ArticleBehaviorMsg msg = new ArticleBehaviorMsg(1640750040932687874L, 2, ArticleBehaviorTypeEnum.VIEWS);
                ProducerRecord<String, String> record1 =
                        new ProducerRecord<String, String>(TOPIC, JSON.toJSONString(msg));
                producer.send(record1);
                log.info("发送数据: {}", msg);

                msg = new ArticleBehaviorMsg(1640750040932687874L, 3, ArticleBehaviorTypeEnum.LIKES);
                record1 = new ProducerRecord<String, String>(TOPIC, JSON.toJSONString(msg));
                producer.send(record1);
                log.info("发送数据: {}", msg);

                msg = new ArticleBehaviorMsg(1640750040932687874L, 4, ArticleBehaviorTypeEnum.COMMENT);
                record1 = new ProducerRecord<String, String>(TOPIC, JSON.toJSONString(msg));
                producer.send(record1);
                log.info("发送数据: {}", msg);

                msg = new ArticleBehaviorMsg(1640750040932687874L, 5, ArticleBehaviorTypeEnum.COLLECTION);
                record1 = new ProducerRecord<String, String>(TOPIC, JSON.toJSONString(msg));
                producer.send(record1);
                log.info("发送数据: {}", msg);

                msg = new ArticleBehaviorMsg(1641701412222013441L, 1, ArticleBehaviorTypeEnum.VIEWS);
                ProducerRecord<String, String> record2 = new ProducerRecord<String, String>(TOPIC, JSON.toJSONString(msg));
                producer.send(record2);
                log.info("发送数据: {}", msg);

                msg = new ArticleBehaviorMsg(1641701412222013441L, 1, ArticleBehaviorTypeEnum.LIKES);
                record2 = new ProducerRecord<String, String>(TOPIC, JSON.toJSONString(msg));
                producer.send(record2);
                log.info("发送数据: {}", msg);

                msg = new ArticleBehaviorMsg(1641702802327711746L, 2, ArticleBehaviorTypeEnum.COMMENT);
                record2 = new ProducerRecord<String, String>(TOPIC, JSON.toJSONString(msg));
                producer.send(record2);
                log.info("发送数据: {}", msg);

                msg = new ArticleBehaviorMsg(1641702802327711746L, 2, ArticleBehaviorTypeEnum.COLLECTION);
                record2 = new ProducerRecord<String, String>(TOPIC, JSON.toJSONString(msg));
                producer.send(record2);
                log.info("发送数据: {}", msg);

            } catch (Exception e) {
                e.printStackTrace();
            }
            //关系消息通道
            producer.close();
            Thread.sleep(100);
        }


    }
}
